package io.jnet.common.businessprocessor;

import java.util.concurrent.ExecutorService;
import com.jfireframework.baseutil.collection.buffer.ByteBuf;
import com.jfireframework.baseutil.concurrent.CpuCachePadingInt;
import com.jfireframework.baseutil.concurrent.SpscQueue;
import io.jnet.common.api.ChannelContext;
import io.jnet.common.api.WriteHandler;
import io.jnet.common.bufstorage.BufStorage;
import io.jnet.common.streamprocessor.ProcesserUtil;
import io.jnet.common.streamprocessor.ProcessorTask;
import io.jnet.common.streamprocessor.StreamProcessor;
import io.jnet.common.util.AioListener;

public class TaskAttachProcessor implements Runnable
{
    private final SpscQueue<ProcessorTask> tasks          = new SpscQueue<>();
    private static final int               IDLE           = 0;
    private static final int               WORK           = 1;
    private final CpuCachePadingInt        status         = new CpuCachePadingInt(IDLE);
    private final ExecutorService          executorService;
    private final StreamProcessor[]        processors;
    private final ChannelContext           serverChannelContext;
    private final AioListener              serverListener;
    private final WriteHandler             writeHandler;
    private final BufStorage               bufStorage;
    private static final int               SPIN_THRESHOLD = 1 << 7;
    private int                            spin           = 0;
    
    public TaskAttachProcessor(ChannelContext serverChannelContext, BufStorage bufStorage, WriteHandler writeHandler, StreamProcessor[] processors, ExecutorService executorService, AioListener serverListener)
    {
        this.executorService = executorService;
        this.serverChannelContext = serverChannelContext;
        this.serverListener = serverListener;
        this.processors = processors;
        this.writeHandler = writeHandler;
        this.bufStorage = bufStorage;
    }
    
    @Override
    public void run()
    {
        do
        {
            ProcessorTask task = tasks.poll();
            if (task == null)
            {
                spin = 0;
                for (;;)
                {
                    
                    if ((task = tasks.poll()) != null)
                    {
                        break;
                    }
                    else if ((spin += 1) < SPIN_THRESHOLD)
                    {
                        ;
                    }
                    else
                    {
                        status.set(IDLE);
                        if (tasks.isEmpty() == false)
                        {
                            tryExecute();
                        }
                        return;
                    }
                }
            }
            try
            {
                Object result = ProcesserUtil.process(serverChannelContext, processors, task.getData(), task.getInitIndex());
                if (result instanceof ByteBuf<?>)
                {
                    bufStorage.putBuf((ByteBuf<?>) result);
                    writeHandler.registerWrite();
                }
            }
            catch (Throwable e)
            {
                serverListener.catchException(e, serverChannelContext);
                writeHandler.registerWrite();
            }
        } while (true);
    }
    
    public void commit(ProcessorTask task)
    {
        tasks.offer(task);
        tryExecute();
    }
    
    private void tryExecute()
    {
        int now = status.value();
        if (now == IDLE && status.compareAndSwap(IDLE, WORK))
        {
            executorService.execute(this);
        }
    }
}
